#define _GNU_SOURCE
#include "memleak/memleak.h"
#include <errno.h>
#include <netinet/in.h>
#include <pthread.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <time.h>
#include <unistd.h>
#include <fcntl.h>
#include "log/log.h"
#include <uv.h>
#include "ringbuf/ringbuf.h"

#ifndef SETNONBLOCK
#define SETNONBLOCK(fd) fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK)
#endif
#define my_calloc(a,b)   ({ int _a = (int)a, _b = (int)b; void *p = calloc(_a, _b); LOGMSG("calling calloc(%u) %p\n", (int)(_a*_b), p); p; })
#define my_free(a)       ({ LOGMSG("calling free(%p)\n", a);     free(a); })
#define uv_close(h,c)    ({ LOGMSG("calling uv_close(%p)\n", h); uv_close(h,c); })
#ifndef container_of
#define container_of(ptr, type, member) ({ typeof(((type *)0)->member) *__mptr = (ptr); (type *)((char *)__mptr - offsetof(type,member));})
#endif


FILE *LLOG_FP = NULL;
LIST_HEAD(mem_node_list);
atomic_t ct_count = ATOMIC_INIT(0);
char *server_host = NULL;
time_t g_startup_time = 0;
int keepalive = 1;
int no_delay = 1;
int verbose = 0;
pthread_mutex_t log_file_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t memleak_lock = PTHREAD_MUTEX_INITIALIZER;
struct log_handle *log_handle = NULL; int log_with_ts = 1, log_with_func = 0, log_with_file = 0, log_with_line = 1;
uint32_t ct_count_max = 0;
unsigned short server_port = 0;
uv_loop_t uv_loop;

struct item {
    int ref;
    int send_req_count;
    uv_connect_t connect;
    uv_tcp_t uv_tcp;
    int uv_tcp_is_connected;
    uv_getaddrinfo_t resolver;
    uv_timer_t timer;
    struct addrinfo hints;
    int is_reading;
};

typedef struct {
    uv_write_t req;
    struct item *item;
    uv_buf_t buf;
} write_req_t;
void on_write(uv_write_t *req, int status);

void alloc_buffer_for_read(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
    static char bbuf[4096];
    buf->base = bbuf;
    buf->len  = sizeof(bbuf);
}

void uv_free_cb(uv_handle_t *handle)
{
    return;
}

void send_data(struct item *item)
{
    if (item->send_req_count > 0) return;
    static char data[1000];
    write_req_t *write_req = (write_req_t*) calloc(1, sizeof(write_req_t));
    item->send_req_count++;
    write_req->buf = uv_buf_init(data, sizeof(data));
    write_req->item = item;
    uv_write(&write_req->req, (uv_stream_t *)&item->uv_tcp, &write_req->buf, 1, on_write);
}

void item_free_cb(uv_handle_t *handle)
{
    uv_tcp_t *uv_tcp = (void *)handle;
    struct item *item = container_of(uv_tcp, struct item, uv_tcp);
    item->ref--;
    if (item->ref <= -2) { free(item); }
}

void timer_free_cb(uv_handle_t *handle)
{
    uv_timer_t *timer = (void *)handle;
    struct item *item = container_of(timer, struct item, timer);
    item->ref--;
    if (item->ref <= -2) { free(item); }
}

void start_one_item(void);
void destory_item(struct item *item)
{
    atomic_dec(&ct_count);
    LOGERR("destroy ct, ct rest %u\n", atomic_read(&ct_count));
    uv_close((uv_handle_t *)&item->timer, timer_free_cb);
    uv_close((uv_handle_t *)&item->uv_tcp, item_free_cb); item->uv_tcp_is_connected = 0;
    start_one_item();
}

void on_read(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf)
{
    uv_tcp_t *uv_tcp = (uv_tcp_t *)client;
    struct item *item = container_of(uv_tcp, struct item, uv_tcp);
    if (nread <= 0) { LOGERR("nread %d error[%s]\n", (int)nread, uv_err_name(nread)); destory_item(item); return; }
    if (item->is_reading) {
        item->is_reading = 0;
        uv_read_stop((uv_stream_t*) &uv_tcp);
    }
}

void on_connected_server(uv_connect_t* req, int status)
{
    struct item *item = container_of(req, struct item, connect);
    if (status) {
        LOGERR("connect server[%s] callback error [%s], ignore this error and go on use old ct\n", server_host, uv_err_name(status));
        destory_item(item);
        return;
    }
    LOGMSG("connect to server ok\n");
    item->uv_tcp_is_connected = 1;
    send_data(item);
    if (item->is_reading == 0) {
	    item->is_reading = 1;
	    uv_read_start((uv_stream_t*) &item->uv_tcp, alloc_buffer_for_read, on_read);
    }
}

#include "host2ip/host2ip.h"
//static inline int host_to_ipstr(const char *host, char *output, int buf_size)
void on_server_host_resolved(uv_getaddrinfo_t *resolver, int status, struct addrinfo *res)
{
    struct item *item = container_of(resolver, struct item, resolver);
    if (status == 0) {
        //char addr[17] = ""; uv_ip4_name((struct sockaddr_in*) res->ai_addr, addr, 16); LLOG("server resolve to %s\n", addr);
        uv_tcp_connect(&item->connect, &item->uv_tcp, (const struct sockaddr*) res->ai_addr, on_connected_server);
        if (keepalive > 0) uv_tcp_keepalive(&item->uv_tcp, 1, 20);
        if (no_delay > 0) uv_tcp_nodelay(&item->uv_tcp, 1);
        //uv_stream_set_blocking((uv_stream_t *)&item->uv_tcp, 0);
    } else {
        LOGERR("server host resolve fail[%s]\n", uv_err_name(status));
        destory_item(item);
    }
    uv_freeaddrinfo(res);
}

void on_tick(uv_timer_t *timer)
{
    struct item *item = container_of(timer, struct item, timer);
    if (item->uv_tcp_is_connected == 1) {
	    send_data(item);
	    if (item->is_reading == 0) {
		    item->is_reading = 1;
		    uv_read_start((uv_stream_t*) &item->uv_tcp, alloc_buffer_for_read, on_read);
	    }
    }
}

void start_one_item(void)
{
    struct item *item = calloc(1, sizeof(*item));
    uv_tcp_init(&uv_loop, &item->uv_tcp);
    item->hints.ai_family = PF_INET;
    item->hints.ai_socktype = SOCK_STREAM;
    item->hints.ai_protocol = IPPROTO_TCP;
    item->hints.ai_flags = 0;
    char portstr[64]; sprintf(portstr, "%u", server_port);
    int r = uv_getaddrinfo(&uv_loop, &item->resolver, on_server_host_resolved, server_host, portstr, &item->hints); if (r) { LOGERR("getaddrinfo call error %s\n", uv_err_name(r)); exit(1); }
    uv_timer_init(&uv_loop, &item->timer);
    uv_timer_start(&item->timer, on_tick, 1000, 1000);
    atomic_inc(&ct_count);
    LOGERR("created one ct. [ct count %u]\n", atomic_read(&ct_count));
}

uint64_t write_total = 0;
void on_write(uv_write_t *req, int status)
{
    write_req_t *write_req = container_of(req, write_req_t, req);
    struct item *item = write_req->item;
    item->send_req_count--;
    if (status) {
	    LOGERR("write error %s\n", uv_err_name(status));
        destory_item(item);
    } else {
        write_total += write_req->buf.len;
        time_t d = time(NULL) - g_startup_time; if (d <= 0) d = 1;
        uint64_t s = write_total / d;
        LOGFEW("write %luB %lus %luBPS\n", write_total, d, s);
	    send_data(item);
    }
    free(write_req);
}

void sig_handle_usr1(int sig)
{
    memleak_stat(NULL, 0);
}

int main(int argc, char **argv)
{
    g_startup_time = time(NULL);
    signal(SIGPIPE, SIG_IGN);
    signal(SIGUSR1, sig_handle_usr1);
    //log_init(0, 0, "/tmp/sender.log", "/tmp/sender.log.2", "/tmp/sender.log.3");
    LLOG("USEAGE: %s -s server_host -p server_port -n binfa_count\n", argv[0]);

    int i;
    unsigned int binfa_count = 1;
    for (i = 1; i < argc; i++) {
        if (0) { }
        else if (strcmp(argv[i], "-s") == 0) { if (i+1 >= argc) { printf("-s need arg!!\n"); exit(1); } server_host = argv[i+1];       i++; }
        else if (strcmp(argv[i], "-p") == 0) { if (i+1 >= argc) { printf("-p need arg!!\n"); exit(1); } server_port = atoi(argv[i+1]); i++; }
        else if (strcmp(argv[i], "-n") == 0) { if (i+1 >= argc) { printf("-n need arg!!\n"); exit(1); } binfa_count = atoi(argv[i+1]); i++; }
        else { printf("unknown arg %s\n", argv[i]); exit(1);                      }
    }
    if (server_host == NULL) { LOGERR("server_host not config. use arg -s xxx\n"); exit(1); }
    if (server_port <= 0)    { LOGERR("server_port not config. use arg -p xxx\n"); exit(1); }
    if (binfa_count <= 0 || binfa_count > 1000)    { LOGERR("bingfa_count %d, should between 1-1000\n", binfa_count); exit(1); }
    LLOG("server_host %s\n", server_host);
    LLOG("server_port %u\n", server_port);
    LLOG("binfa_count %u\n", binfa_count);
    uv_loop_init(&uv_loop);

    for (i = 0; i < binfa_count; i++) {
        start_one_item();
    }

    return uv_run(&uv_loop, UV_RUN_DEFAULT);
}
